//
// Created by cube on 2020/9/12.
//

#ifndef COMPILERDOMEA1_CRUDP_H
#define COMPILERDOMEA1_CRUDP_H

#include <iostream>
#include <map>

#include "../lib/Base/Object.h"

#include "CommonNetBase/Buffer.h"
#include "Option.h"
#include "Peer.h"
#include "UdpPool.h"
#include "Agreement.h"

using namespace NetBase;
using namespace NetBaseCommon;

class CRUDPA : public Lib::EventDriver,protected UdpPool {
protected:
    LIB_EVENT_(Server_UdpLoop_Event)
public:
    CRUDPA(const int count) : UdpPool(count) {
        setLoop(&__loop);
    }

    void bind(const NetAddress &addr) { UdpPool::bind(addr); }

    void ubind(const NetAddress &addr) { UdpPool::ubind(addr); }

    uint32_t send(const NetAddress &addr, int8_t *data, const uint32_t &len) {
        TMutex<Peer> *peer = getPeer(addr);
        uint32_t rul = peer->lock()->send(data, len);
        send_frame_to_io(addr);
        return rul;
    }

    uint32_t recv(const NetAddress &addr, int8_t *data, const uint32_t &len) {
        TMutex<Peer> *peer = nullptr;
        if (true) {
            auto l = __hosts.lock_guard();
            if (!__hosts.contains(addr))return 0;
            peer = __hosts.at(addr);
        }
        return peer->lock()->recv(data, len);
    }

    void setRecv(std::function<void(const NetAddress& addr)> recv){__recv = recv;}
    void setTimeOut(std::function<void(const NetAddress& addr)> timeout){__timeout = timeout;}

    const NetBaseCommon::Option &option() { return __option; }

protected:
    TMutex<Peer> *getPeer(const NetAddress &addr) {
        auto l = __hosts.lock_guard();
        if (!__hosts.contains(addr)) {
            auto peer = new TMutex<Peer>(__option);
            peer->setLoop(loop());
            __hosts.insert(std::pair(addr, peer));
        }
        return __hosts.at(addr);
    }

    void send_frame_to_io(const NetAddress &addr) {
        TMutex<Peer> *peer = getPeer(addr);
        if(!peer->mutex().try_lock())return;
        std::lock_guard l(peer->mutex(),std::adopt_lock);
        while (peer->canSend() && !peer->empty()) {
            Frame *frame = new Frame();
            frame->data = new int8_t[__option.__cmd_buffer_size];

            bool no_only = false;
            uint32_t size = write_to_frame(peer, frame->data, __option.__cmd_buffer_size,no_only);

            if(size == 0)return;
            frame->len = size;
            if (no_only)peer->sendFrame(frame, size);
            else peer->sendFrameOnly(frame, size);
        }
    }

    int read_from_io(NetUdp &udp, NetAddress &addr) override {
        std::cout<<"recv from io"<<addr.getIp_Char()<<std::endl;
        int8_t buf[1024];
        int rul = udp.recv(addr, buf, sizeof(buf));
        if (rul > 0) {
            bool recv_buf = false;
            if(true){
                auto peer = getPeer(addr);
                auto l = peer->lock_guard();
                auto size = (peer->__send_buffer != nullptr) ? peer->__send_buffer->size() : 0;
                read_from_io(peer, buf, rul);
                if(peer->__send_buffer != nullptr)
                    if(size != peer->__send_buffer->size())recv_buf = true;
            }
            if(recv_buf)__recv(addr);
            send_frame_to_io(addr);
        }
        return rul;
    }

    virtual uint32_t write_to_frame(TMutex<Peer> *peer, int8_t *data,const uint32_t &size,bool& no_only) {
        if (size < sizeof(CommandHead))return 0;
        auto &head = *(CommandHead *) data;
        uint32_t cmd_size = 0;
        no_only = write_command_parse(peer, data + sizeof(CommandHead), size - sizeof(CommandHead),cmd_size);
        if(cmd_size == 0)return 0;
        return cmd_size + sizeof(CommandHead);
    }

    virtual void read_from_io(TMutex<Peer> *peer, int8_t *data, const uint32_t &size) {
        if (size < sizeof(CommandHead))return;
        auto &head = *(CommandHead *) data;
        bool re_ack = false;
        read_command_parse(peer, data + sizeof(CommandHead), size - sizeof(CommandHead),re_ack);
        if(re_ack)peer->__acks.push(head.id);
    }

    void read_command_parse(TMutex<Peer> *peer, int8_t *data, const uint32_t &size,bool& re_ack) {
        auto& head = *(CommandHead *) (data - sizeof(CommandHead));
        auto& acks = peer->__acks;
        auto& cmds = peer->__cmds;
        auto& sendbuf = peer->__send_buffer;
        auto& cmdbase = *(CommandBase*)data;
        switch (cmdbase.type){
            case CommandType::ReAck_Type:{
                if(size < sizeof(ReAck))break;
                auto& cmd = *(ReAck*)data;
                auto now = getNow();
                auto rtt = now - head.time;
                peer->onAck(cmd.id);
                peer->service(rtt);
                read_command_parse(peer,data + sizeof(ReAck),size - sizeof(ReAck),re_ack);
                break;
            }case CommandType::ReAcks_Type:{
                if(size < sizeof(ReAcks))break;
                auto& cmd = *(ReAcks*)data;
                auto* cmd_data = data + sizeof(ReAcks);
                auto now = getNow();
                auto rtt = now - head.time;
                for(int i = 0;i<cmd.count;i++)
                    peer->onAck(*(((uint16_t*)cmd_data)+i));
                peer->service(rtt);
                re_ack = true;
                auto index_len = sizeof(ReAcks) + cmd.count * sizeof(uint16_t);
                read_command_parse(peer,data + index_len,size - index_len,re_ack);
                break;
//            }case CommandType::Start_Type:{
//                if(size < sizeof(Start_Command))break;
//                auto& cmd = *(Start_Command*)data;
//                re_ack = true;
//                read_command_parse(peer,data + sizeof(Start_Command),size - sizeof(Start_Command),re_ack);
//                break;
//            }case CommandType::Stop_Type:{
//                if(size < sizeof(Stop_Command))break;
//                auto& cmd = *(Stop_Command*)data;
//                re_ack = true;
//                read_command_parse(peer,data + sizeof(Stop_Command),size - sizeof(Stop_Command),re_ack);
//                break;
            }case CommandType::SendData_Type:{
                if(size < sizeof(SendData))break;
                auto& cmd = *(SendData*)data;
                auto* cmd_data = data + sizeof(SendData);
                auto* f = new Frame();
                f->data = new int8_t[cmd.len]{};
                f->len = cmd.len;
                memcpy(f->data,cmd_data,cmd.len);
                peer->writeToBuffer(head.id,f);
                re_ack = true;
                auto index_len = sizeof(SendData) + cmd.len;
                read_command_parse(peer,data + index_len,size - index_len,re_ack);
                break;
            }default:{
                return;
            }
        }
    }

    bool write_command_parse(TMutex<Peer> *peer, int8_t *data, const uint32_t &size,uint32_t& cmd_size) {
        auto& acks = peer->__acks;
        auto& cmds = peer->__cmds;
        auto& sendbuf = peer->__send_buffer;
        if(!acks.empty()){
            if (size > sizeof(ReAck) * 3 && acks.size() > 3) {
                auto &cmd = *(ReAcks *) data;
                uint32_t _size = sizeof(ReAcks);
                auto* cmd_data_p = data + sizeof(ReAcks);
                cmd.type = CommandType::ReAcks_Type;
                cmd.count = 0;
                while (!acks.empty() && (size - _size) > sizeof(Frame_Id)){
                    *(((Frame_Id*)cmd_data_p)+cmd.count) = acks.front();
                    _size += sizeof(Frame_Id);
                    acks.pop();
                    cmd.count++;
                }
                write_command_parse(peer,data + _size,size - _size,cmd_size += _size);
            } else if (size > sizeof(ReAck)) {
                const auto& cmd_s = sizeof(ReAck);
                auto &cmd = *(ReAck *) data;
                cmd.type = CommandType::ReAck_Type;
                return write_command_parse(peer,data + cmd_s,size - cmd_s,cmd_size += cmd_s);
            }
        } else if (!cmds.empty()) {
            while (!cmds.empty()){
//                auto cmd_p = cmds.front();
//                if(cmd_p->type == CommandType::Stop_Type){
//                    const auto &cmd_s = sizeof(Stop_Command);
//                    auto &cmd = *(Stop_Command *) data;
//                    cmd.type = CommandType::Stop_Type;
//                    write_command_parse(peer, data + cmd_s, size - cmd_s, cmd_size += cmd_s);
//                }
//                cmds.pop();
            }
        } else if(sendbuf != nullptr && size > sizeof(SendData) && !sendbuf->empty()){
            const auto &cmd_s = sizeof(SendData);
            auto &cmd = *(SendData *) data;
            cmd.type = CommandType::SendData_Type;
            cmd_size += sendbuf->read(data + cmd_s, size);
        } else {
          return false;
        }
        return true;
    }

protected:
    void event(EventAgent& event) override {
        if(event->getType() == Server_UdpLoop_Event::type()){
            std::cout<<"sss"<<std::endl;
            this->server(__option.__recv_time.count());
            event.reSendTo(__option.__recv_interval_time,loop());
        }else if(event->getType() == Peer::SendToIO_Event::type()){
            auto _eve = (Peer::SendToIO_Event*)event.data();
            send_to_io(_eve->addr,_eve->frame);
        }else if(event->getType() == Peer::SendAckToIO_Event::type()){
            auto _eve = (Peer::SendAckToIO_Event*)event.data();
            send_to_io(_eve->addr,_eve->frame,_eve->id);
        }else if(event->getType() == Peer::SendTimeOut_Event::type()) {
            auto _eve = (Peer::SendTimeOut_Event*)event.data();
            sendTimeout(_eve->addr);
        }else if(event->getType() == StartEvent::type()){
            sendEvent(new Server_UdpLoop_Event());
        }
    }

    virtual void send_to_io(const NetAddress& addr,Frame* frame){
        std::cout<<"send to io"<<std::endl;
        UdpPool::send(addr,frame->data,frame->len);
        send_frame_to_io(addr);
        delete frame;
    }
    virtual void send_to_io(const NetAddress& addr,Frame* frame,const Frame_Id& id){
        std::cout<<"send to io "<<id<<std::endl;
        ((CommandHead*)frame->data)->id = id;
        ((CommandHead*)frame->data)->time = getNow();
        UdpPool::send(addr,frame->data,frame->len);
        send_frame_to_io(addr);
    }
    void sendTimeout(const NetAddress& addr){
        auto l = __hosts.lock_guard();
        if(__hosts.contains(addr)){
            TMutex<Peer>* p =__hosts.at(addr);
            delete p;
            __hosts.erase(addr);
            __timeout(addr);
        }
    }

protected:
    NetBaseCommon::Option __option;
    EventLoop             __loop;
    TMap<NetAddress,TMutex<Peer>*>   __hosts;
    std::function<void(const NetAddress& addr)> __timeout;
    std::function<void(const NetAddress& addr)> __recv;
};
#endif //COMPILERDOMEA1_CRUDP_H
